﻿#region Copyright (c) 2013-04, Olaf Kober <amarok.projects@gmail.com>
//================================================================================
//	Permission is hereby granted, free of charge, to any person obtaining a copy
//	of this software and associated documentation files (the "Software"), to deal
//	in the Software without restriction, including without limitation the rights
//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//	copies of the Software, and to permit persons to whom the Software is
//	furnished to do so, subject to the following conditions:
//
//	The above copyright notice and this permission notice shall be included in
//	all copies or substantial portions of the Software.
//
//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//	THE SOFTWARE.
//================================================================================
#endregion

#if TESTS

using System;
using System.Threading;
using NUnit.Framework;


namespace Amarok.Agents
{
	[TestFixture]
	public class Test_DefaultPublisher
	{
		private class TestMessage : Message
		{
		}

		private class TestInformation : Information
		{
		}

		private class StubAgentSubscriber : Agent
		{
			public static Int32 HandlerCalled;
			public volatile Boolean IsTornDown;

			public StubAgentSubscriber(AgentEnvironment environment, AgentOptions options)
				: base(environment, options)
			{
			}

			public void TearDown()
			{
				IsTornDown = true;
			}

			[MessageHandler, ExclusiveExecution, NoParallelismAttribute]
			private void _HandleMessage(Message message)
			{
				if (IsTornDown)
					return;

				HandlerCalled++;
			}
		}

		private class StubAgentSubscriber2 : Agent
		{
			public static Int32 HandlerCalled;
			public volatile Boolean IsTornDown;

			public StubAgentSubscriber2(AgentEnvironment environment, AgentOptions options)
				: base(environment, options)
			{
			}

			public void TearDown()
			{
				IsTornDown = true;
			}

			[MessageHandler, ExclusiveExecution, NoParallelismAttribute]
			private void _HandleMessage(Message message)
			{
				if (IsTornDown)
					return;

				HandlerCalled++;
			}
		}


		[SetUp]
		public void Setup()
		{
			StubAgentSubscriber.HandlerCalled = 0;
			StubAgentSubscriber2.HandlerCalled = 0;
		}


		// --- TESTS ---


		[Test]
		public void Test_Publish()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(TestMessage),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();

			}
		}

		[Test]
		public void Test_Publish_MultipleMessages()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(Message),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				for (Int32 i = 0; i < 100000; i++)
					publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 100000);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(100000));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_SubscribedForBaseMessageType()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(Message),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());
				publisher.Publish(new TestInformation());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 2);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(2));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_NotSubscribedForMessageType()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(TestInformation),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 0);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_WithMessageFilter()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(Message),
					subscriber.ToAgentReference(),
					m => m is TestInformation);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				var message = new TestMessage();
				publisher.Publish(message);

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 0);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				var information = new TestInformation();
				publisher.Publish(information);

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_NullMessage()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(TestMessage),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				Assert.Throws<ArgumentNullException>(() =>
					{
						publisher.Publish(null);
					});

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 0);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_AfterComplete()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(TestMessage),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));

				publisher.Complete();
				publisher.Completion.Wait();

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Subscribe_NullMessageType()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				Assert.Throws<ArgumentNullException>(() =>
					{
						publisher.Subscribe(null, subscriber.ToAgentReference(), null);
					});
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Subscribe_NonMessageType()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				Assert.Throws<ArgumentException>(() =>
					{
						publisher.Subscribe(typeof(Version), subscriber.ToAgentReference(), null);
					});
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Subscribe_NullSubscriber()
		{
			var publisher = new DefaultPublisher();

			Assert.Throws<ArgumentNullException>(() =>
				{
					publisher.Subscribe(typeof(Message), (AgentReference)null, null);
				});
		}

		[Test]
		public void Test_Publish_AfterSubscriptionDisposed()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			try
			{
				var subscription = publisher.Subscribe(
					typeof(TestMessage),
					subscriber.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));

				subscription.Dispose();
				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber);
				subscriber.TearDown();
			}
		}

		[Test]
		public void Test_Publish_AfterSubscriberGarbageCollected()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			var subscription = publisher.Subscribe(
				typeof(TestMessage),
				subscriber.ToAgentReference(),
				null);

			Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));

			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
			Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));

			GC.KeepAlive(subscriber);

			subscriber = null;
			GC.Collect(2, GCCollectionMode.Forced);
			GC.WaitForPendingFinalizers();

			publisher.Publish(new TestMessage());
			publisher.Publish(new TestMessage());
			publisher.Publish(new TestMessage());
			subscription.Completion.Wait(5000);

			Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_After_Complete()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber = new StubAgentSubscriber(environment, AgentOptions.Empty);

			publisher.Complete();
			publisher.Completion.Wait(5000);

			var subscription = publisher.Subscribe(
				typeof(TestMessage),
				subscriber.ToAgentReference(),
				null);

			Assert.That(subscription, Is.Null);
		}

		[Test]
		public void Test_Publish_MultipleSubscribers()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber1 = new StubAgentSubscriber(environment, AgentOptions.Empty);
			var subscriber2 = new StubAgentSubscriber2(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(Message),
					subscriber1.ToAgentReference(),
					null);
				publisher.Subscribe(
					typeof(Message),
					subscriber2.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber2.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber1);
				GC.KeepAlive(subscriber2);

				subscriber1.TearDown();
				subscriber2.TearDown();
			}
		}

		[Test]
		public void Test_Publish_MultipleSubscribers_ManyMessages()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber1 = new StubAgentSubscriber(environment, AgentOptions.Empty);
			var subscriber2 = new StubAgentSubscriber2(environment, AgentOptions.Empty);

			try
			{
				publisher.Subscribe(
					typeof(Message),
					subscriber1.ToAgentReference(),
					null);
				publisher.Subscribe(
					typeof(Message),
					subscriber2.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(0));

				for (Int32 i = 0; i < 100000; i++)
					publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 100000);
				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber2.HandlerCalled == 100000);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(100000));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(100000));
			}
			finally
			{
				GC.KeepAlive(subscriber1);
				GC.KeepAlive(subscriber2);

				subscriber1.TearDown();
				subscriber2.TearDown();
			}
		}

		[Test]
		public void Test_Publish_MultipleSubscribers_DisposingSubscriptions()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber1 = new StubAgentSubscriber(environment, AgentOptions.Empty);
			var subscriber2 = new StubAgentSubscriber2(environment, AgentOptions.Empty);

			try
			{
				var subscription1 = publisher.Subscribe(
					typeof(Message),
					subscriber1.ToAgentReference(),
					null);
				var subscription2 = publisher.Subscribe(
					typeof(Message),
					subscriber2.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber2.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(1));

				subscription1.Dispose();
				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 2);
				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber2.HandlerCalled == 2);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(2));

				subscription2.Dispose();
				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 3);
				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber2.HandlerCalled == 3);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(2));
			}
			finally
			{
				GC.KeepAlive(subscriber1);
				GC.KeepAlive(subscriber2);

				subscriber1.TearDown();
				subscriber2.TearDown();
			}
		}

		[Test]
		public void Test_Publish_MultipleSubscribers_CompletingSubscriptions()
		{
			var publisher = new DefaultPublisher();
			var environment = new AgentEnvironment();
			var subscriber1 = new StubAgentSubscriber(environment, AgentOptions.Empty);
			var subscriber2 = new StubAgentSubscriber2(environment, AgentOptions.Empty);

			try
			{
				var subscription1 = publisher.Subscribe(
					typeof(Message),
					subscriber1.ToAgentReference(),
					null);
				var subscription2 = publisher.Subscribe(
					typeof(Message),
					subscriber2.ToAgentReference(),
					null);

				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(0));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(0));

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber.HandlerCalled == 1);
				Helper.WaitForCondition(5000, 10, () => StubAgentSubscriber2.HandlerCalled == 1);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(1));

				subscription1.Complete();
				subscription2.Complete();
				Assert.That(TaskHelper.WhenAll(subscription1.Completion, subscription2.Completion).Wait(5000), Is.True);

				publisher.Publish(new TestMessage());

				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber.HandlerCalled == 2);
				Helper.WaitForCondition(500, 10, () => StubAgentSubscriber2.HandlerCalled == 2);
				Assert.That(StubAgentSubscriber.HandlerCalled, Is.EqualTo(1));
				Assert.That(StubAgentSubscriber2.HandlerCalled, Is.EqualTo(1));
			}
			finally
			{
				GC.KeepAlive(subscriber1);
				GC.KeepAlive(subscriber2);

				subscriber1.TearDown();
				subscriber2.TearDown();
			}
		}

		[Test]
		public void Test_Publish_Callback()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			publisher.Subscribe(
				typeof(TestMessage),
				callback,
				null);

			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(5000, 10, () => callbackCalled == 1);
			Assert.That(callbackCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Publish_Callback_SubscribedForBaseMessageType()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			publisher.Publish(new TestMessage());
			publisher.Publish(new TestInformation());

			Helper.WaitForCondition(5000, 10, () => callbackCalled == 2);
			Assert.That(callbackCalled, Is.EqualTo(2));
		}

		[Test]
		public void Test_Publish_Callback_NotSubscribedForMessageType()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			publisher.Subscribe(
				typeof(TestInformation),
				callback,
				null);

			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(500, 10, () => callbackCalled == 0);
			Assert.That(callbackCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Publish_Callback_WithMessageFilter()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			publisher.Subscribe(
				typeof(TestInformation),
				callback,
				m => m is TestInformation);

			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(500, 10, () => callbackCalled == 0);
			Assert.That(callbackCalled, Is.EqualTo(0));

			publisher.Publish(new TestInformation());

			Helper.WaitForCondition(500, 10, () => callbackCalled == 1);
			Assert.That(callbackCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Callback_NullMessageType()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			Assert.Throws<ArgumentNullException>(() =>
				{
					publisher.Subscribe(null, callback, null);
				});
		}

		[Test]
		public void Test_Subscribe_Callback_NonMessageType()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				callbackCalled++;

			Assert.Throws<ArgumentException>(() =>
				{
					publisher.Subscribe(typeof(Version), callback, null);
				});
		}

		[Test]
		public void Test_Subscribe_Callback_NullCallback()
		{
			var publisher = new DefaultPublisher();

			Assert.Throws<ArgumentNullException>(() =>
				{
					publisher.Subscribe(typeof(TestInformation), (Action<Message>)null, null);
				});
		}

		[Test]
		public void Test_Publish_Callback_Disposed()
		{
			var publisher = new DefaultPublisher();

			Int32 callback1Called = 0;
			Action<Message> callback1 = msg =>
				callback1Called++;

			Int32 callback2Called = 0;
			Action<Message> callback2 = msg =>
				callback2Called++;

			var token1 = publisher.Subscribe(
				typeof(TestMessage),
				callback1,
				null);
			var token2 = publisher.Subscribe(
				typeof(TestMessage),
				callback2,
				null);

			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(5000, 10, () => callback1Called == 1);
			Assert.That(callback1Called, Is.EqualTo(1));
			Helper.WaitForCondition(5000, 10, () => callback2Called == 1);
			Assert.That(callback2Called, Is.EqualTo(1));

			token1.Dispose();
			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(500, 10, () => callback1Called == 2);
			Assert.That(callback1Called, Is.EqualTo(1));
			Helper.WaitForCondition(5000, 10, () => callback2Called == 2);
			Assert.That(callback2Called, Is.EqualTo(2));

			token2.Dispose();
			publisher.Publish(new TestMessage());

			Helper.WaitForCondition(500, 10, () => callback1Called == 3);
			Assert.That(callback1Called, Is.EqualTo(1));
			Helper.WaitForCondition(500, 10, () => callback2Called == 3);
			Assert.That(callback2Called, Is.EqualTo(2));
		}

		[Test]
		public void Test_Complete_ProcessPendingMessages()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				{
					SpinWait.SpinUntil(() => false, 1000);
					callbackCalled++;
				};

			var subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			Assert.That(callbackCalled, Is.EqualTo(0));


			// publish messages
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait until first message is being processed
			Thread.Sleep(250);

			// complete while first message is being processed
			publisher.Complete(false);
			publisher.Completion.Wait();

			// all messages should have been processed
			Assert.That(callbackCalled, Is.EqualTo(5));

			// publishing new messages has no effect
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait some more time; ensure that no other messages have been processed
			Thread.Sleep(2000);
			Assert.That(callbackCalled, Is.EqualTo(5));

			// completing twice has no effect
			publisher.Complete(true);
			publisher.Completion.Wait();

			// disposing the subscription after completion has no effect
			subscription.Dispose();

			// subscribing after completion has no effect
			subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);
			Assert.That(subscription, Is.Null);
		}

		[Test]
		public void Test_Complete_DiscardPendingMessages()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				{
					SpinWait.SpinUntil(() => false, 1000);
					callbackCalled++;
				};

			var subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			Assert.That(callbackCalled, Is.EqualTo(0));


			// publish messages
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait until first message is being processed
			Thread.Sleep(250);

			// complete while first message is being processed
			publisher.Complete(true);
			publisher.Completion.Wait();

			// only the first message should have been processed
			Assert.That(callbackCalled, Is.EqualTo(1));

			// publishing new messages has no effect
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait some more time; ensure that no other messages have been processed
			Thread.Sleep(2000);
			Assert.That(callbackCalled, Is.EqualTo(1));

			// completing twice has no effect
			publisher.Complete(true);
			publisher.Completion.Wait();

			// disposing the subscription after completion has no effect
			subscription.Dispose();

			// subscribing after completion has no effect
			subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);
			Assert.That(subscription, Is.Null);
		}

		[Test]
		public void Test_Dispose_DiscardPendingMessages()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
			{
				SpinWait.SpinUntil(() => false, 1000);
				callbackCalled++;
			};

			var subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			Assert.That(callbackCalled, Is.EqualTo(0));


			// publish messages
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait until first message is being processed
			Thread.Sleep(250);

			// complete while first message is being processed
			Helper.Dispose(publisher);

			// only the first message should have been processed
			Assert.That(callbackCalled, Is.EqualTo(1));

			// publishing new messages has no effect
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait some more time; ensure that no other messages have been processed
			Thread.Sleep(2000);
			Assert.That(callbackCalled, Is.EqualTo(1));

			// completing twice has no effect
			Helper.Dispose(publisher);

			// disposing the subscription after completion has no effect
			subscription.Dispose();

			// subscribing after completion has no effect
			subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);
			Assert.That(subscription, Is.Null);
		}

		[Test]
		public void Test_DisposeSubscription()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
				{
					SpinWait.SpinUntil(() => false, 1000);
					callbackCalled++;
				};

			var subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			Assert.That(callbackCalled, Is.EqualTo(0));


			// publish messages
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait until first message is being processed
			Thread.Sleep(250);

			// dispose subscription while first message is being processed
			subscription.Dispose();

			// publishing new messages has no effect
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// all messages should have been processed
			Thread.Sleep(7000);
			Assert.That(callbackCalled, Is.EqualTo(1));

			// disposing the subscription twice has no effect
			subscription.Dispose();
		}

		[Test]
		public void Test_CompleteSubscription_ProcessPendingMessages()
		{
			var publisher = new DefaultPublisher();

			Int32 callbackCalled = 0;
			Action<Message> callback = msg =>
			{
				SpinWait.SpinUntil(() => false, 1000);
				callbackCalled++;
			};

			var subscription = publisher.Subscribe(
				typeof(Message),
				callback,
				null);

			Assert.That(callbackCalled, Is.EqualTo(0));


			// publish messages
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// wait until first message is being processed
			Thread.Sleep(250);

			// complete subscription while first message is being processed
			subscription.Complete(discardAvailableMessages: false);

			// publishing new messages has no effect
			for (Int32 i = 0; i < 5; i++)
				publisher.Publish(new TestMessage());

			// all messages should have been processed
			Assert.That(subscription.Completion.Wait(7000), Is.True);
			Assert.That(callbackCalled, Is.EqualTo(5));
		}

	}
}

#endif
